package com.honeybee.cloud.framework.rocket.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.util.Objects;

/**
 * @Description 一些mq推送基础的消费者
 * @Author DELL
 * @Create 2019-04-03
 * @Since 1.0.0
 */
@Slf4j
public abstract class AbstractPushConsumer implements InitializingBean {

    /**
     * 消费者组
     */
    protected String consumerGroup = "DEFAULT_CONSUMER_GROUP";

    /**
     * mq连接路径
     */
    protected String nameSrvAddress;

    /**
     * tags
     */
    protected String tags = "*";

    /**
     * topic
     */
    protected String topic;

    /**
     * ConsumeFromWhere
     */
    protected ConsumeFromWhere consumeFromWhere;


    /**
     * 启动监听
     *
     * @return
     */
    protected abstract MessageListener listener();

    /**
     * 初始化
     *
     * @throws Exception
     */
    @Override
    public abstract void afterPropertiesSet() throws Exception;

    /**
     * 使用时请调用此方法实例化，完成相关配置
     *
     * @param nameSrvAddress
     * @param topic
     * @param tags
     * @param consumerGroup
     */
    protected void init(String nameSrvAddress, String topic, String tags, String consumerGroup) {
        init(nameSrvAddress, topic, tags, consumerGroup, null);
    }

    /**
     * 使用时请调用此方法实例化，完成相关配置
     *
     * @param nameSrvAddress
     * @param topic
     * @param tags
     * @param consumerGroup
     */
    protected void init(String nameSrvAddress, String topic, String tags, String consumerGroup, ConsumeFromWhere consumeFromWhere) {
        this.nameSrvAddress = nameSrvAddress;
        this.topic = topic;
        this.tags = tags;
        this.consumerGroup = consumerGroup;
        this.consumeFromWhere = consumeFromWhere;
    }

    /**
     * 开启消费者
     *
     * @throws MQClientException
     */
    protected void startConsumerGroup() throws MQClientException {
        log.info("开始启动消费组:{},地址:{}", consumerGroup, nameSrvAddress);
        Assert.hasLength(nameSrvAddress, "nameSrvAddress is not null");
        Assert.hasLength(topic, "topic is not null");
        Assert.hasLength(tags, "tags is not null");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameSrvAddress);
        consumer.setConsumeFromWhere(Objects.isNull(consumeFromWhere) ? ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET : consumeFromWhere);
        StringBuffer subExpression = new StringBuffer();
        subExpression.append(tags);
        consumer.subscribe(topic, subExpression.toString());
        consumer.setMessageListener(listener());
        consumer.setVipChannelEnabled(false);
        consumer.start();
        log.info("成功启动消费组:{}", consumerGroup);
    }


}
